package org.zjt.chapter1.rxnetty.loadbalance;

import io.netty.buffer.ByteBuf;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.concurrent.GlobalEventExecutor;
import io.reactivex.netty.client.Host;
import io.reactivex.netty.client.events.ClientEventListener;
import io.reactivex.netty.client.loadbalancer.AbstractP2CStrategy;
import io.reactivex.netty.client.loadbalancer.LoadBalancerFactory;
import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.events.HttpClientEventsListener;
import io.reactivex.netty.protocol.http.client.loadbalancer.EWMABasedP2CStrategy;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import io.reactivex.netty.protocol.http.client.HttpClient;
import rx.Observable;
import rx.observables.BlockingObservable;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URL;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import static org.zjt.chapter1.rxnetty.RxNettyTest.getByteBuf2StringFunc;
/**
 * Description:
 *
 *          ChannelGroup channel管理器。实现对所有channel 数据写出、关闭。
 *
 *
 * @author juntao.zhang
 * Date: 2018-07-22 下午12:14
 * @see
 */
@Slf4j
public class LoadBalanceClientTest {



    @Test
    public void client() {

        ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

        channels.writeAndFlush("ssssssssssss");


        Observable<Host> hosts = Observable.from(Arrays.asList("localhost:8080", "localhost:8081"))
                .map(item -> new InetSocketAddress(item.split(":")[0], Integer.valueOf(item.split(":")[1])))
                .map(item -> new Host(item));


        HttpClientRequest<ByteBuf, ByteBuf> httpClientRequest = HttpClient.newClient(LoadBalancerFactory.create(new MyLoadBalancer()), hosts)
                .intercept().next(provider -> (version, method, uri) -> provider.createRequest(version, method, uri)
                        .addHeader("url", uri)).finish()
                .createGet("localhost:8080");

        BlockingObservable<Observable<String>> blockingObservable = httpClientRequest.map(resp -> resp.getContent().map(getByteBuf2StringFunc())).toBlocking();

        blockingObservable.subscribe(content -> content.subscribe(log::info));
    }


    /**
     * 按照权重的大小，来实现负载均衡。
     */
    public static final class MyLoadBalancer extends AbstractP2CStrategy {

        /**
         * 返回一个负载均衡的监听器，监听当前的请求的事件。
         * @param host
         * @return
         */
        @Override
        protected ClientEventListener newListener(Host host) {
            return new ClientEventListenerImpl();
        }

        /**
         * 利用当前的服务均衡监听器来得到当前host的负载均衡的信息。
         * @param clientEventListener
         * @return
         */
        @Override
        protected double getWeight(ClientEventListener clientEventListener) {
            return ((ClientEventListenerImpl)clientEventListener).weight;
        }


        /**
         * 负载均衡策略：
         *
         *      每个服务器的服务的权重设置为最高，当服务器内部出现内部异常时候，就会将权重减一。
         */
        @Slf4j
        public static class ClientEventListenerImpl extends HttpClientEventsListener {

            private volatile long weight = Long.MAX_VALUE;

            @Override
            public void onConnectFailed(long duration, TimeUnit timeUnit, Throwable throwable) {
                weight = Long.MIN_VALUE;
            }

            @Override
            public void onByteRead(long bytesRead) {
                super.onByteRead(bytesRead);
            }

            @Override
            public void onResponseHeadersReceived(int responseCode, long duration, TimeUnit timeUnit) {
                log.error("Response code: " + responseCode);
                if (HttpResponseStatus.SERVICE_UNAVAILABLE.code() == responseCode) {
                    weight = Long.MIN_VALUE;
                }
            }
        }
    }
}
